package org.devgframwork.datasync.core.support.executors.source;

import com.alibaba.fastjson.JSONObject;
import org.devgframwork.datasync.core.entity.ScheduleConfig;
import org.devgframwork.datasync.core.entity.TargetConfig;
import org.devgframwork.datasync.core.model.ExecutorModel;
import org.devgframwork.datasync.core.model.ResultModel;
import org.devgframwork.datasync.core.support.DataSourceMemory;
import org.devgframwork.datasync.core.support.executors.AbstractExecutor;
import org.devgframwork.datasync.core.support.executors.ExecutorCore;
import org.devgframwork.datasync.core.support.parser.HandleParser;
import org.devgframwork.datasync.core.support.parser.impl.SqlDefaultParser;
import org.devgframwork.datasync.core.support.service.ScheduleConfigService;
import org.devgframwork.datasync.core.utils.CommonUtil;
import org.devgframwork.datasync.core.utils.SearchUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

public class JdbcMySqlExecutor implements AbstractExecutor<ScheduleConfig> {

    private static Logger log = LoggerFactory.getLogger(JdbcMySqlExecutor.class);

    private List<TargetConfig> targetConfigList;

    public JdbcMySqlExecutor(List<TargetConfig> targetConfigList) {
        this.targetConfigList = targetConfigList;
    }

    @Override
    public ResultModel invoke(ExecutorModel<ScheduleConfig> executorModel) {
        ScheduleConfig scheduleConfig = executorModel.getExt();
        ResultModel preCheck = preCheck(scheduleConfig, log);
        if (!preCheck.isSuccess()) {
            return preCheck;
        }
        String sourceKey = scheduleConfig.getSourceKey();
        String sourceContent = scheduleConfig.getSourceContent();
        if (!DataSourceMemory.jdbcTemplateMap.containsKey(sourceKey)) {
            return ResultModel.newFail("sync taskName:" + scheduleConfig.getTaskName() + " sourceKey:" + sourceKey + " 不存在!");
        }
        if (!DataSourceMemory.sqlMap.containsKey(sourceContent)) {
            return ResultModel.newFail("sync taskName:" + scheduleConfig.getTaskName() + " sourceContent:" + sourceContent + " 不存在!");
        }
        String maxTrackColumnValue = null;
        long t1 = System.currentTimeMillis();
        ScheduleConfigService scheduleConfigService = DataSourceMemory.applicationContext.getBean(ScheduleConfigService.class);
        ThreadPoolExecutor threadPoolExecutor = null;
        try {
            Integer batchSize = (scheduleConfig.getBatchSize() == null ? DataSourceMemory.scheduleBatchSize : scheduleConfig.getBatchSize());
            JdbcTemplate sourceJdbcTemplate = DataSourceMemory.jdbcTemplateMap.get(scheduleConfig.getSourceKey());
            String sourceContentSql = DataSourceMemory.sqlMap.get(scheduleConfig.getSourceContent());
            Integer isTrack = scheduleConfig.getIsTrack();
            String trackColumnName = scheduleConfig.getTrackColumnName();
            Map<String, Object> queryMap = new HashMap<>();
            if (null != isTrack && isTrack.equals(1) && !CommonUtil.isEmpty(trackColumnName)) {
                String trackColumnValue = scheduleConfig.getTrackColumnValue();
                queryMap.put(SearchUtil.lastSqlValue, trackColumnValue);
            }
            HandleParser handleParser = new SqlDefaultParser(sourceContentSql, queryMap);
            String sqlText = handleParser.getText();
            String sqlCount = "select count(*) from (" + sqlText + ") _sqlCount";
            int count = sourceJdbcTemplate.queryForObject(sqlCount, Integer.class);
            int pageNum = count % batchSize == 0 ? count / batchSize : count / batchSize + 1;
            // 判断最大更新值
            maxTrackColumnValue = getMaxTrackColumnValue(sourceJdbcTemplate, sqlText, scheduleConfig);
            // 更新同步数量
            updateSyncNumber(scheduleConfigService, scheduleConfig.getScheduleConfigId(), count);
            // 更新数据
            Integer treadNum = (scheduleConfig.getThreadNum() == null ? 1 : scheduleConfig.getThreadNum());
            threadPoolExecutor = new ThreadPoolExecutor(
                    treadNum,
                    treadNum,
                    30L,
                    TimeUnit.SECONDS,
                    new LinkedBlockingQueue<Runnable>(1000),
                    new ThreadFactory() {
                        @Override
                        public Thread newThread(Runnable r) {
                            return new Thread(r, "data-sync, admin JdbcToJdbcExecutor-threadPoolExecutorThreadPool-" + r.hashCode());
                        }
                    },
                    new RejectedExecutionHandler() {
                        @Override
                        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                            r.run();
                            log.warn(">>>>>>>>>>> data-sync, registry or remove too fast, match threadpool rejected handler(run now).");
                        }
                    });
            Map<String, Future<ResultModel>> futureMap = new HashMap<>();
            for (int page = 0; page < pageNum; page++) {
                ExecutorCore executorCore = new ExecutorCore(page, batchSize, sqlText, sourceJdbcTemplate, scheduleConfigService, scheduleConfig, targetConfigList, queryMap);
                Future<ResultModel> future = threadPoolExecutor.submit(executorCore);
                String key = scheduleConfig.getTaskName() + "_" + page;
                futureMap.put(key, future);
            }
            for (String key : futureMap.keySet()) {
                Future<ResultModel> future = futureMap.get(key);
                ResultModel resultModel = future.get();
                log.info("JdbcMySqlExecutor invoke key:{},resultModel:{}", key, JSONObject.toJSONString(resultModel));
            }
        } catch (Exception ex) {
            log.error("JdbcToJdbcExecutor error", ex.getMessage());
            ex.printStackTrace();
            return ResultModel.newFail(ex.getMessage());
        } finally {
            if (null != threadPoolExecutor) {
                threadPoolExecutor.shutdown();
            }
            realseRunning(scheduleConfigService, scheduleConfig.getScheduleConfigId(), t1, maxTrackColumnValue);
        }
        return ResultModel.newSuccess();
    }
}
